package edu.xidian.sei.mlcs

import scala.collection.mutable.ListBuffer
import edu.xidian.sei.mlcs.util.Stopwatch
import java.util.concurrent.ForkJoinPool
import java.util.concurrent.ForkJoinTask
import java.util.concurrent.RecursiveTask
import akka.actor.Actor
import akka.pattern.ask
import scala.concurrent.Await
import akka.actor.ActorSystem
import akka.util.Timeout
import scala.concurrent.Await
import scala.concurrent.duration.DurationInt
import com.carrotsearch.sizeof.RamUsageEstimator
import scala.collection.mutable.ArrayBuffer

class Crawler {

}

object Crawler {

  val actorNum = 5
  val actorList = new ArrayBuffer[SubGraphActor]

  def main(args: Array[String]): Unit = {
    if (args.length == 0) {
      println(
        "Usage:HorizontalCrawler /path/to/your/data/file -Dmlcs.max-thread=2 -Dmlcs.auto-shrink=true -Dmlcs.optimal-only=false");
      return ;
    }

    val prompts = new StringBuilder

    val mlcs = MLCS.loadFromFile(args(0))
    mlcs.maxThread = 2

    val maxThread = System.getProperty("mlcs.max-thread")
    if (null != maxThread) {
      mlcs.maxThread = Integer.parseInt(maxThread);
      prompts.append(" max-thread=" + maxThread);
    }

    println("Starting search " + args(0) + prompts)

    val startTime = System.nanoTime()
    val graph = new Graph(mlcs)
    val result = bfs(mlcs, graph)
    System.out.println(result.count + " mlcs(max length " + result.length + ") found, using "
      + Stopwatch.format(System.nanoTime() - startTime));
    if (null != result.count) {
      val total = result.count.intValue();
      if (total > 0 && total <= 100)
        graph.printPathResult;
    }
    println(RamUsageEstimator.sizeOf(mlcs))
    println(RamUsageEstimator.sizeOf(graph))
  }

  def bfs(mlcs: MLCS, graph: Graph): Result = {
    val start = mlcs.start;
    val heads = new ListBuffer[Location]();
    mlcs.nextSortedLocations(start).foreach { loc =>
      if (graph.addEdge(null, loc))
        heads += loc
    }

    search(mlcs, graph, heads);
    return graph.stat;
  }

  def search(mlcs: MLCS, graph: Graph, heads: ListBuffer[Location]) {
    val pool = if (mlcs.maxThread > 0) new ForkJoinPool(mlcs.maxThread) else new ForkJoinPool()
    val levelQueue = new ListBuffer[Location]()
    levelQueue ++= heads
    val parallelism = pool.getParallelism();
    val startTime = System.nanoTime()
    while (!levelQueue.isEmpty) {
      val countPerSlice = Math.ceil(levelQueue.size * 1.0 / parallelism).toInt
      val locsList = levelQueue.sliding(countPerSlice, countPerSlice)
      val tasks = new ListBuffer[ForkJoinTask[ListBuffer[Location]]]
      locsList.foreach { locs =>
        tasks += pool.submit(new BFSCrawler(mlcs, locs))
      }
      levelQueue.clear
      tasks.foreach { task =>
        val nextLocs = task.join()
        nextLocs.foreach { loc => if (graph.addEdge(null, loc)) levelQueue += loc }
      }

    }
    pool.shutdown
    println("build graph using " + Stopwatch.format(System.nanoTime() - startTime))
  }

}

//class BFSCrawler(mlcs: MLCS, locations: ListBuffer[Location]) extends RecursiveTask[ListBuffer[Location]] {
//  def compute(): ListBuffer[Location] = {
//    val nexts = new ListBuffer[Location] //下次拓扑排序的点的层和点集
//    locations.foreach { loc => //当前点的loc
//      nexts ++= mlcs.nextLocations(loc)
//    }
//    nexts //返回一个层和改层点的映射
//  }
//}
